package com.suning.yace;

import java.util.ArrayList;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

import org.apache.http.HttpEntity;
import org.apache.http.NameValuePair;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;

import com.alibaba.fastjson.JSONObject;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;

public class Connection {

    private static final int HEARTBEAT_INTERVAL = 300;

    @SuppressWarnings("rawtypes")
	private final ChannelInitializer channelInitializer = new DeviceChannelInitializer();

    private final ChannelHandler deviceHandler = new DeviceHandler();

    private final EventLoopGroup eventLoopGroup;

    private Channel channel;

    private String deviceId;

    private String nodeHost;

    private int nodePort;

    private String handshakeToken;

    private long lastHeartbeatSendTime = 0;

    private Queue<String> failMessages = new ConcurrentLinkedQueue<String>();

    public Connection(String deviceId, final EventLoopGroup eventLoopGroup) {
        this.deviceId = deviceId;
        this.eventLoopGroup = eventLoopGroup;
    }

    public void start() throws Exception {
        selectNode();
        connect();
    }

    public void checkHeartbeat() {
        long time = System.currentTimeMillis() - lastHeartbeatSendTime;
        if (lastHeartbeatSendTime != 0 && time > (HEARTBEAT_INTERVAL + 10) * 1000) {
            System.out.println("device " + deviceId + " not send heartbeat in time," +
                    "last send time is " + time / 1000 + " s ago.");
        }
    }

    private Bootstrap configureBootstrap(EventLoopGroup eventLoopGroup) {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.remoteAddress(nodeHost, nodePort);
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        bootstrap.group(eventLoopGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(channelInitializer);
        return bootstrap;
    }

    public void connect() {
        channel = configureBootstrap(eventLoopGroup).connect().channel();
    }

    private void doHandshake(String token) {
        sendMessage(RequestUtils.getHandShakeRequest(deviceId, deviceId,  HEARTBEAT_INTERVAL, token).toJSONString(), false);
    }

    public void sendMessage(final String message, final boolean retry) {
        channel.writeAndFlush(message).addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture future) {
                if (!future.isSuccess()) {
                    if (retry) {
                        failMessages.add(message);
                    }
                }
            }
        });
    }

    private void selectNode() {
        try {
        	CloseableHttpClient httpClient = HttpClients.createDefault();
            HttpPost httppost = new HttpPost("http://lvcpre.cnsuning.com/lvcenter/handshake.do");  
            List<NameValuePair> formparams = new ArrayList<NameValuePair>();  
            formparams.add(new BasicNameValuePair("custNum", deviceId));  
            formparams.add(new BasicNameValuePair("nickName", deviceId));  
            formparams.add(new BasicNameValuePair("headImg", deviceId));  
            UrlEncodedFormEntity uefEntity = new UrlEncodedFormEntity(formparams, "UTF-8");  
            httppost.setEntity(uefEntity);
            CloseableHttpResponse response = httpClient.execute(httppost);
            HttpEntity entity = response.getEntity(); 
            try {
				String jsonStr = EntityUtils.toString(entity, "UTF-8");
				JSONObject json = JSONObject.parseObject(jsonStr);
				if (200 == json.getInteger("code").intValue()) {
					JSONObject data = json.getJSONObject("data");
					this.nodeHost = "10.37.193.39";
					this.nodePort = data.getInteger("port").intValue();
					this.handshakeToken = data.getString("token");
				} else {
				    throw new Exception("selectNode get empty response.");
				}
			} finally {
				response.close();
			}
        } catch (Exception ex) {
            throw new RuntimeException("selectNode exception:", ex);
        }
    }

    final LiveLengthFieldPrepender liveLengthFieldEncoder = new LiveLengthFieldPrepender();
    private class DeviceChannelInitializer extends ChannelInitializer<Channel> {

        protected void initChannel(Channel channel) throws Exception {
            ChannelPipeline p = channel.pipeline();
            p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 2, 4, 0, 6));
            p.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
            p.addLast("frameEncoder", liveLengthFieldEncoder);
            p.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
            p.addLast("handler", deviceHandler);
        }
    }

    @ChannelHandler.Sharable
    private class DeviceHandler extends SimpleChannelInboundHandler<String> {

        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        	System.out.println(deviceId + " connected to: " + ctx.channel().remoteAddress());
            doHandshake(handshakeToken);
        }

        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
            System.out.println(deviceId + " received Message \r\n" + msg);
            JSONObject jsonObj = JSONObject.parseObject(msg);
            String type = jsonObj.getString("type");
            if (type.equals(ResponseUtils.HSR)) {
                handleHandshakeResponse(msg, ctx);
            } else if (type.equals(ResponseUtils.Recived.Push2ChatRoom)) {
                handleServerPush(msg, ctx);
            } else if (type.equals(ResponseUtils.HBR)) {
                handleHeartbeatResponse(msg, ctx);
            }
        }

        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        	System.out.println("Got disconnected... will try to reconnect in 1 sec...");
            final EventLoop loop = ctx.channel().eventLoop();
            loop.schedule(new Runnable() {
                public void run() {
                	System.out.println("Reconnecting...");
                    connect();
                }
            }, 1, TimeUnit.SECONDS);
        }

        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        	System.out.println("Unexpected exception from downstream." + cause);
            ctx.close();
        }

        private void handleHandshakeResponse(String msg, ChannelHandlerContext ctx) {
            JSONObject jsonObj = JSONObject.parseObject(msg);
            int code = jsonObj.getInteger(ResponseUtils.CODE).intValue();
            if (code == 200) {
            	System.out.println(deviceId + " handshake success");
                if (!failMessages.isEmpty()) {
                    while (true) {
                        try {
                            sendMessage(failMessages.remove(), true);
                        } catch (NoSuchElementException ex) {
                            break;
                        }
                    }
                }
                sendMessage(RequestUtils.getJoinRoomRequest("lyRoom").toJSONString(), true);
                eventLoopGroup.schedule(new Runnable() {
                    public void run() {
                        sendMessage(RequestUtils.getHeartBeatRequest().toJSONString(), false);
                        lastHeartbeatSendTime = System.currentTimeMillis();
                    }
                }, HEARTBEAT_INTERVAL, TimeUnit.SECONDS);
            } else {
            	System.out.println(deviceId + " handshake failed");
                selectNode();
                ctx.close();
            }
        }

        private void handleServerPush(String msg, ChannelHandlerContext ctx) {
                System.out.println(deviceId + " receive push message \r\n" + msg);
        }

        private void handleHeartbeatResponse(String msg, ChannelHandlerContext ctx) {
        	System.out.println(deviceId + " get heartbeat response.");
            eventLoopGroup.schedule(new Runnable() {
                public void run() {
                    sendMessage(RequestUtils.getHeartBeatRequest().toJSONString(), false);
                    lastHeartbeatSendTime = System.currentTimeMillis();
                }
            }, HEARTBEAT_INTERVAL, TimeUnit.SECONDS);
        }
    }    
}
